package com.lagou.kafka.Kafka;


import com.lagou.kafka.entity.Model;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpRequest;
import org.apache.http.params.HttpParams;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.*;

import javax.servlet.http.HttpServletRequest;
import java.util.Date;
import java.util.concurrent.ExecutionException;


/**
 * @author: yehw
 * @date: 2020/8/19 01:06
 * @description: 消息生产者
 */
@RestController
@Slf4j
public class ProducerMessage {
    Logger logger = LoggerFactory.getLogger(ProducerMessage.class);
    @Autowired()
    KafkaTemplate kafkaTemplate;
    @GetMapping("/hello")
    public String index(){
        return "hello";
    }
    @PostMapping("/nginx")
    public String get(HttpServletRequest request) throws ExecutionException, InterruptedException {
        logger.info("开始选择参数");
        String act_time =(String) request.getParameter("act_time");
        String action =(String) request.getParameter("action");
        String job_code =(String) request.getParameter("job_code");
        String user_id =(String) request.getParameter("user_id");
        Model message=new Model();
        message.setAct_time(new Date());
        message.setAction(action);
        message.setJob_code(job_code);
        message.setUser_id(user_id);
        ListenableFuture send = kafkaTemplate.send(
                new ProducerRecord<Integer, String>(
                        "topic-nginx",
                        0,
                        1,
                        message.toString()));
        Object o = send.get();
        SendResult<Integer, String> sendResult = (SendResult<Integer, String>) o;
        System.out.println(sendResult.getRecordMetadata().topic() + sendResult.getRecordMetadata().partition() + sendResult.getRecordMetadata().offset());
        return "success";
    }
}
